package cn.com.bluemoon.bd.flink.func.filter;

import cn.com.bluemoon.bd.flink.creater.state.ValueStateCreater;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

/**
 * 根据字段来过滤，实现distinct操作
 */
public class DistinctFilterByField extends RichFilterFunction<String> {
    // 状态名称
    private String stateName;
    // 状态TTL
    private Time stateTTL;
    // 状态存储
    private ValueState<Boolean> state;

    public DistinctFilterByField(String iStateName) {
        this(iStateName, Time.days(1));
    }

    public DistinctFilterByField(String iStateName, Time iStateTTL) {
        this.stateName = iStateName;
        this.stateTTL = iStateTTL;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.state = ValueStateCreater.create(this.stateName, this.stateTTL, boolean.class, this.getRuntimeContext());
    }

    @Override
    public boolean filter(String s) throws Exception {
        Boolean value = state.value();
        if (value == null) {
            state.update(true);
            return true;
        } else {
            return false;
        }
    }

}
